AWS CloudTrailの証跡情報をAmazon Athenaで集計してみた

武田隆志

2017.12.21

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

こんにちは。サービスグループの武田です。

今回はAWS CloudTrailで集積した証跡情報をAmazon Athenaで集計する方法を紹介します。

CloudTrailのログからAthenaでレポートを作成する方法は、以前にも石川さんや西澤さんが紹介していますので、併せてご覧ください。

Amazon Athena CloudTrailSerdeを利用してAWSのアクティビティを探索する

CloudTrailログからAmazon Athenaを使ってログイン監査レポートを作成する

前提

事前にCloudTrailの設定を行い、Amazon S3に証跡情報が出力されている必要があります。

なお、CloudTrailは各リージョンごとに有効化する方法と全リージョン有効化する方法の2種類提供されていますが、どちらの設定でも同じように集計できます。

全リージョンで有効化していた場合は、リージョンを横断して集計できるので複数リージョンを利用している場合は少し便利です。

環境

今回は次の環境で実行しています。

macOS Sierra
AWS CLI
jq

$ aws --version
aws-cli/1.14.0 Python/3.6.3 Darwin/16.7.0 botocore/1.8.4
$ jq --version
jq-1.5

テーブル作成

まずはAthenaのクエリを発行するためにテーブルを作成します。

次の4つのパラメータは自身の環境に合わせて、適宜変更してください。

ATHENA_TABLE_NAME（任意のテーブル名）
SOURCE_BUCKET（CloudTrailの集積用バケット）
OUTPUT_BUCKET（結果格納用バケット）
AWS_ACCOUNT_ID（自身のAWSアカウントID）

$ ATHENA_TABLE_NAME=cloudtrail_all_region
$ SOURCE_BUCKET=cloudtail-all-region-123456789012
$ OUTPUT_BUCKET=aws-athena-query-result-123456789012-20171221
$ AWS_ACCOUNT_ID=123456789012

$ aws athena start-query-execution --result-configuration OutputLocation="s3://$OUTPUT_BUCKET" --query-string \
"CREATE EXTERNAL TABLE IF NOT EXISTS $ATHENA_TABLE_NAME (
awsregion STRING,
eventsource STRING,
eventname STRING,
eventtime STRING
) PARTITIONED BY (
region string,
year string,
month string,
day string
)
ROW FORMAT SERDE 'com.amazon.emr.hive.serde.CloudTrailSerde'
STORED AS INPUTFORMAT 'com.amazon.emr.cloudtrail.CloudTrailInputFormat'
OUTPUTFORMAT 'org.apache.hadoop.hive.ql.io.HiveIgnoreKeyTextOutputFormat'
LOCATION 's3://$SOURCE_BUCKET/AWSLogs/$AWS_ACCOUNT_ID/CloudTrail/'"

クエリの実行結果（ステータス）を確認したい場合は、実行結果として出力されるQueryExecutionIdを、aws athena get-query-executionで指定すれば確認できます。

マネジメントコンソールでも確認してみると、きちんとテーブルが作成されています。

しかし、このままクエリを発行するとすべてのデータをスキャンしてしまい、時間とお金がかかってしまいます。パーティションの設定をして特定日付のみ効率よく集計できるようにします。

S3バケットにはCloudTrail/region/year/month/dayという階層でログが保管されています。そのため、全リージョンで有効化している場合は、リージョンごとにパーティションを作成します。単一リージョンの場合は、パーティションも1つで十分です。

次の3つのパラメータは集計対象の年月日に合わせて、適宜変更してください。この例では2012年12月01日のログを集計するためのパーティションを作成しています。

PART_YEAR（年）
PART_MONTH（月）
PART_DAY（日）

$ PART_YEAR=2017
$ PART_MONTH=12
$ PART_DAY=01

$ for region in $(aws ec2 describe-regions --query 'Regions[].RegionName' | jq -r '.[]'); do \
aws athena start-query-execution --result-configuration OutputLocation="s3://$OUTPUT_BUCKET" --query-string \
"ALTER TABLE $ATHENA_TABLE_NAME ADD PARTITION (region='$region',year='$PART_YEAR',month='$PART_MONTH',day='$PART_DAY') LOCATION 's3://$SOURCE_BUCKET/AWSLogs/$AWS_ACCOUNT_ID/CloudTrail/$region/$PART_YEAR/$PART_MONTH/$PART_DAY/'"; \
done

集計クエリの発行

それでは先ほど作成したテーブルとパーティションを利用してクエリを発行しましょう。

リージョンごと、イベントソース/イベントごとにイベントの回数を集計し、回数を降順でソートします。ソートの順番はORDER BY句で好きに指定できるため、自身の欲しい結果となるように指定すればOKです。

$ qid=$(aws athena start-query-execution --result-configuration OutputLocation="s3://$OUTPUT_BUCKET" --query-string \
"SELECT
awsregion,
eventsource,
eventname,
date(from_iso8601_timestamp(eventtime)) AS eventtime,
COUNT(*) AS count
FROM $ATHENA_TABLE_NAME
WHERE
year = '$PART_YEAR' AND month = '$PART_MONTH' AND day = '$PART_DAY'
GROUP BY
awsregion,
eventsource,
eventname,
date(from_iso8601_timestamp(eventtime))
ORDER BY count DESC" \
| jq -r '.QueryExecutionId')

$ aws s3 cp s3://$OUTPUT_BUCKET/$qid.csv cloudtrail-$PART_YEAR$PART_MONTH$PART_DAY.csv

あとはダウンロードできたCSVファイルをExcelなり、別のプログラムに食わせるなりして、好きに利用できます。